package com.atiguigu.rabbitmq.four;

import com.atiguigu.rabbitmq.utils.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

/*
* 发布确认模式：使用时间，比较那种确认模式最好
* 1.单个确认
* 2.批量确认
* 异步批量确认
* */
public class ConfirmMessage {
    public static  final int MESSAGE_COUNT=1000;
    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        //1.单个确认   1000条消息单个确认耗费1487ms
        //publishMessageIndividually();

        //* 2.批量确认 1000条消息批量确认耗费400ms
        //publishMessageBatch();
        //* 异步批量确认 发布1000条消息异步发布确认时间：274ms
        publishMessageAsync();
    }

    //单个确认
    public static  void  publishMessageIndividually() throws IOException, TimeoutException, InterruptedException {
        Channel channel = RabbitMqUtils.getChannel();
        String queueName= UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //start time
        long start=System.currentTimeMillis();
        //批量发消息
        for(int i=0; i<MESSAGE_COUNT; i++){
            String message=i+" ";
            channel.basicPublish("",queueName,null,message.getBytes());
            //单个消息马上发布确认
            boolean flag = channel.waitForConfirms();
            if(flag){
                System.out.println("消息发送成功");
            }

        }
        long end= System.currentTimeMillis();
        System.out.println("发布"+MESSAGE_COUNT+"条消息单独确认时间："+(end-start)+"ms");
    }


    //* 2.批量确认
    public static  void  publishMessageBatch() throws IOException, TimeoutException, InterruptedException {
        Channel channel = RabbitMqUtils.getChannel();
        String queueName= UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //start time
        long start=System.currentTimeMillis();
        //批量确认的消息量
        int batchSize = 100;
        //批量发消息_批量发布确认
        for(int i=0; i<MESSAGE_COUNT; i++){
            String message=i+" ";
            channel.basicPublish("",queueName,null,message.getBytes());
            //达到100条消息批量确认一次
            if(i % batchSize == 0){
                boolean flag = channel.waitForConfirms();
                if(flag){
                    System.out.println("消息发送成功");
                }
            }
        }
        long end= System.currentTimeMillis();
        System.out.println("发布"+MESSAGE_COUNT+"条消息批量确认时间："+(end-start)+"ms");
    }


    //* 异步批量确认
    public static  void  publishMessageAsync() throws IOException, TimeoutException, InterruptedException {
        Channel channel = RabbitMqUtils.getChannel();
        String queueName= UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();

        /*
        * 线程安全有序的哈希表，使用高并发的情况,消息缓存队列，用于后续异步确认
        * 1.将序号与消息关联
        * 2.可批量删除条目，只要给到序号
        * 3.支持高并发（多线程）
        * */
        ConcurrentSkipListMap<Long,String> outstandingConfirms = new ConcurrentSkipListMap<>();

        //start time
        long start=System.currentTimeMillis();

        //消息确认成功回调
        ConfirmCallback ackCallback=(deliveryTag,multiple)->{
            //处理异步未确认消息：第2步 删除已经确认的消息,剩下的消息就是未确认的消息
            if(multiple){
                //若是批量确认消息
                //返回的是小于等于当前序列号的未确认消息 是一个 map
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag);
                //清除该部分未确认消息
                confirmed.clear();
            }else{
                //只清除当前序列号的消息
                outstandingConfirms.remove(deliveryTag);
            }


            System.out.println("确认的消息："+deliveryTag);

        };
        //消息确认失败回调 参数1：消息标记 参数2：是否批量确认
        ConfirmCallback nackCallback=(deliveryTag,multiple)->{
            //处理异步未确认消息：第3步打印未确认的消息都有哪
            String message= outstandingConfirms.get(deliveryTag);
            System.out.println("未确认的消息："+message);
        };
        //准备消息监听器 监听：参数1那些消息成功了 参数2那些消息失败了
        channel.addConfirmListener(ackCallback,nackCallback);//异步通知
        int batchSize = 100;
        //批量发消息_异步批量发布确认
        for(int i=0; i<MESSAGE_COUNT; i++){
            String message=i+" ";
            channel.basicPublish("",queueName,null,message.getBytes());
            //处理异步未确认消息：第一步 此处记录下所有要发布的消息
            /**
             * channel.getNextPublishSeqNo()获取下一个消息的序列号
             * 通过序列号与消息体进行一个关联
             * 全部都是未确认的消息体
             */
             outstandingConfirms.put(channel.getNextPublishSeqNo(),message);
        }

        long end= System.currentTimeMillis();
        System.out.println("发布"+MESSAGE_COUNT+"条消息异步发布确认时间："+(end-start)+"ms");
    }
}
